/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.operation;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.DatasetFactory;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.results.Results;
import cgl.granules.results.ResultsFactory;
import cgl.granules.util.MarshallingUtil;
import cgl.narada.util.UUIDRetriever;

/**
 * This is the base class which will serve as the base for implementations of
 * the <code>Map</code> and <code>Reduce</code> interface.
 * 
 * @author Shrideep Pallickara
 * 
 */
public abstract class MapReduceBase extends OperationBase implements Map,
		Reduce {

	private ConcurrentHashMap<String, Reduce> reducers;
	private ConcurrentHashMap<String, Map> mappers;
	private ConcurrentHashMap<String, Integer> resultsTracker;

	private String mrDatasetIdentifier = "MapReduce Dataset";
	private StreamingAccess mapReduceDataset;
	private ResultsFactory resultsFactory = null;
	private UUIDRetriever retriever;
	private String defaultOutputStream = null;
	private int defaultSynopsisType = StreamingAccess.STRING_SYNOPSIS;
	private String moduleName = "MapReduceBase: ";

	public MapReduceBase() {

		resultsFactory = ResultsFactory.getInstance();
		initializeDataStructures();
	}

	@Override
	public DatasetCollection getDatasetCollection() {
		if (!hasDatasetCollection()) {
			initializeDatasetCollection();
		}

		return super.getDatasetCollection();
	}

	protected StreamingAccess getMapReduceDataset() {
		if (mapReduceDataset == null) {
			initializeDatasetCollection();
		}
		return mapReduceDataset;
	}

	private void initializeDataStructures() {
		reducers = new ConcurrentHashMap<String, Reduce>();
		mappers = new ConcurrentHashMap<String, Map>();
		resultsTracker = new ConcurrentHashMap<String, Integer>();
		/** Initialize data collection */
	}

	private void initializeDatasetCollection() {
		if (hasDatasetCollection()) {
			// System.out.println(
			// "Not performing DatasetCollection initializations");
			return;
		}

		DatasetFactory datasetFactory = DatasetFactory.getInstance();
		DatasetCollection datasetCollection = datasetFactory
				.createDatasetCollection();

		String mrDatasetDescription = "Dataset for Map-Reduce Communications";
		mapReduceDataset = datasetFactory.createStreamingDataset(
				mrDatasetIdentifier, mrDatasetDescription);
		setDatasetCollection(datasetCollection);

		retriever = UUIDRetriever.getInstance();
		defaultOutputStream = retriever.getRandomBasedUUIDAsString();

		try {
			datasetCollection.addDataset(mapReduceDataset);

			mapReduceDataset.addOutputStream(defaultOutputStream,
					defaultSynopsisType);
		} catch (DatasetException e) {
			System.out.println(moduleName + "Problems initializing the "
					+ "MapReduce dataset. This SHOULD NOT HAPPEN!"
					+ e.getMessage());
		}
	}

	@Override
	public int addReduce(Reduce reduce) throws MapReduceException {
		initializeDatasetCollection();
		String errorReport = null;
		if (reduce == null) {
			errorReport = "The specified Reduce instance is NULL.";
			throw new NullPointerException(errorReport);
		}

		String reduceIdentifier = reduce.getInstanceIdentifier();
		if (reducers.containsKey(reduceIdentifier)) {
			errorReport = "The specified Reduce instance with id:"
					+ reduceIdentifier + ", has been registered previously";
			throwMapReduceException(errorReport);
		}

		StreamingAccess reduceDataset = ((MapReduceBase) reduce)
				.getMapReduceDataset();

		try {
			reduceDataset.addInputStreams(mapReduceDataset.getOutputStreams());
		} catch (DatasetException e) {
			throwMapReduceException(e.getMessage());
		}

		((MapReduceBase) reduce).addMapper(this);

		reducers.put(reduceIdentifier, reduce);
		return reducers.size();
	}

	@Override
	public int removeReduce(String reduceIdentifier) throws MapReduceException {
		initializeDatasetCollection();
		String errorReport = null;
		if (reduceIdentifier == null) {
			errorReport = "The specified reduceIdentifier is NULL.";
			throw new NullPointerException(errorReport);
		}

		if (!reducers.containsKey(reduceIdentifier)) {
			errorReport = "The specified Reduce instance with id:"
					+ reduceIdentifier + ", has NOT been registered previously";
			throwMapReduceException(errorReport);
		}

		Reduce reduce = reducers.get(reduceIdentifier);
		StreamingAccess reduceDataset = ((MapReduceBase) reduce)
				.getMapReduceDataset();

		try {
			reduceDataset.removeInputStreams(mapReduceDataset
					.getOutputStreams());
		} catch (DatasetException e) {
			throwMapReduceException(e.getMessage());
		}

		((MapReduceBase) reduce).removeMapper(this);

		reducers.remove(reduceIdentifier);
		return reducers.size();
	}

	@Override
	public int getNumberOfReducers() {
		return reducers.size();
	}

	@Override
	public Enumeration<String> getReduceIdentifiers() {
		return reducers.keys();
	}

	@Override
	public int addMap(Map map) throws MapReduceException {
		initializeDatasetCollection();
		String errorReport = null;
		if (map == null) {
			errorReport = "The specified Map instance is NULL.";
			throw new NullPointerException(errorReport);
		}

		String mapIdentifier = map.getInstanceIdentifier();
		if (mappers.containsKey(mapIdentifier)) {
			errorReport = "The specified Map instance with id:" + mapIdentifier
					+ ", has been registered previously";
			throwMapReduceException(errorReport);
		}

		StreamingAccess mapDataset = ((MapReduceBase) map)
				.getMapReduceDataset();

		try {
			mapReduceDataset.addInputStreams(mapDataset.getOutputStreams());
		} catch (DatasetException e) {
			throwMapReduceException(e.getMessage());
		}

		((MapReduceBase) map).addReducer(this);
		mappers.put(mapIdentifier, map);

		return mappers.size();
	}

	@Override
	public int removeMap(String mapIdentifier) throws MapReduceException {
		initializeDatasetCollection();
		String errorReport = null;
		if (mapIdentifier == null) {
			errorReport = "The specified Map identifier is NULL.";
			throw new NullPointerException(errorReport);
		}

		if (!mappers.containsKey(mapIdentifier)) {
			errorReport = "The specified Map instance with id:" + mapIdentifier
					+ ", has NOT been registered previously";
			throwMapReduceException(errorReport);
		}

		Map map = mappers.get(mapIdentifier);

		StreamingAccess mapDataset = ((MapReduceBase) map)
				.getMapReduceDataset();

		try {
			mapReduceDataset.removeInputStreams(mapDataset.getOutputStreams());
		} catch (DatasetException e) {
			throwMapReduceException(e.getMessage());
		}

		((MapReduceBase) map).removeReducer(this);

		mappers.remove(mapIdentifier);
		return mappers.size();
	}

	@Override
	public Enumeration<String> getMapIdentifiers() {
		return mappers.keys();
	}

	@Override
	public int getNumberOfMaps() {
		return mappers.size();
	}

	@Override
	public boolean isAwaitingOutputs() {
		if (resultsTracker.containsValue(0)) {
			return true;
		}

		return false;
	}

	@Override
	public void setReceivedOutputFrom(String mapIdentifier) {
		Integer value = resultsTracker.get(mapIdentifier);
		if (value == null) {
			System.out.println("Received output from unknown map: "
					+ mapIdentifier);
			return;
		}
		value++;
		resultsTracker.put(mapIdentifier, value);
	}

	@Override
	public Iterator<String> getAwaitingOutputsFrom() {
		ConcurrentLinkedQueue<String> awaitingOutputsFrom = new ConcurrentLinkedQueue<String>();
		Enumeration<String> mapKeys = resultsTracker.keys();
		while (mapKeys.hasMoreElements()) {
			String mapId = mapKeys.nextElement();

			if (resultsTracker.get(mapId) == 0) {
				awaitingOutputsFrom.add(mapId);
			}
		}

		Iterator<String> iterator = awaitingOutputsFrom.iterator();
		return iterator;
	}

	
	@Override
	public void resetOutputTracker() {
		Enumeration<String> mapKeys = resultsTracker.keys();
		while (mapKeys.hasMoreElements()) {
			String mapId = mapKeys.nextElement();
			resultsTracker.put(mapId, 0);
		}
	}
	
	/**
	 * This method is called on a Map, once it has been added to a
	 * <code>Reduce</code> using the Reduce.addMap() method.
	 * 
	 * @param reduce
	 */
	protected void addReducer(Reduce reduce) throws MapReduceException {
		String reduceIdentifier = reduce.getInstanceIdentifier();
		if (reducers.containsKey(reduceIdentifier)) {
			String errorReport = "The specified Reduce instance with id:"
					+ reduceIdentifier
					+ ", has been registered previously with the Map: "
					+ getInstanceIdentifier();
			throwMapReduceException(errorReport);
		}
		reducers.put(reduceIdentifier, reduce);
	}

	/**
	 * This method is called on a Map, once it has been removed from a
	 * <code>Reduce</code> using the Reduce.removeMap() method.
	 * 
	 * @param reduce
	 */
	protected void removeReducer(Reduce reduce) throws MapReduceException {
		String reduceIdentifier = reduce.getInstanceIdentifier();
		if (!reducers.containsKey(reduceIdentifier)) {
			String errorReport = "The specified Reduce instance with id:"
					+ reduceIdentifier
					+ ", has NOT been registered previously with the Map: "
					+ getInstanceIdentifier();
			throwMapReduceException(errorReport);
		}
		reducers.remove(reduceIdentifier);
	}

	/**
	 * This method is called on a Reduce, once it has been added to a
	 * <code>Map</code> using the Map.addReduce() method.
	 * 
	 * @param map
	 */
	protected void addMapper(Map map) throws MapReduceException {
		String mapIdentifier = map.getInstanceIdentifier();
		if (reducers.containsKey(mapIdentifier)) {
			String errorReport = "The specified Map instance with id:"
					+ mapIdentifier
					+ ", has been registered previously with the Reduce: "
					+ getInstanceIdentifier();
			throwMapReduceException(errorReport);
		}
		mappers.put(mapIdentifier, map);
	}

	/**
	 * This method is called on a Reduce, once it has been removed from a
	 * <code>Map</code> using the Map.removeReduce() method.
	 * 
	 * @param map
	 */
	protected void removeMapper(Map map) throws MapReduceException {
		String mapIdentifier = map.getInstanceIdentifier();
		if (!reducers.containsKey(mapIdentifier)) {
			String errorReport = "The specified Map instance with id:"
					+ mapIdentifier
					+ ", has NOT been registered previously with the Reduce: "
					+ getInstanceIdentifier();
			throwMapReduceException(errorReport);
		}
		mappers.remove(mapIdentifier);
	}

	@Override
	public Results createResults(boolean completeResult,
			boolean processingComplete) {
		Results results = resultsFactory.createResult(getInstanceIdentifier(),
				completeResult, processingComplete);
		return results;
	}

	@Override
	public void writeResults(Results results) throws MapReduceException {
		writeResults(defaultOutputStream, results);
	}

	@Override
	public void writeResults(String streamSynopsis, Results results)
			throws MapReduceException {
		writeResults(defaultSynopsisType, streamSynopsis, results);
	}
	
	@Override
	public void writeResults(int synopsisType, String streamSynopsis, 
			Results results) throws MapReduceException {
		try {
			byte[] payload = results.getBytes();
			if (mapReduceDataset == null) {
				System.out.println("MapReduceDataset is null.");
			}
			mapReduceDataset.writeStreamingData(synopsisType,
					streamSynopsis, payload);
			// System.out.println("Wrote results on: " + streamSynopsis);
		} catch (IOException e) {
			String errorReport = "Problems marshalling results"
					+ e.getMessage();
			throwMapReduceException(errorReport);
		} catch (DatasetException e) {
			String errorReport = "Problems streaming the results"
					+ e.getMessage();
			throwMapReduceException(errorReport);
		}
	}

	@Override
	public void unmarshall(byte[] marshalledBytes) throws IOException,
			DatasetException {
		MarshallingUtil marshaller = MarshallingUtil.getInstance();
		ByteArrayInputStream baInputStream = new ByteArrayInputStream(
				marshalledBytes);
		DataInputStream din = new DataInputStream(new BufferedInputStream(
				baInputStream));

		byte[] baseClassBytes = new byte[din.readInt()];
		din.readFully(baseClassBytes);
		super.unmarshall(baseClassBytes);

		mapReduceDataset = (StreamingAccess) getDatasetCollection().getDataset(
				mrDatasetIdentifier);
		defaultOutputStream = marshaller.readString(din);
		// System.out.println("Default Output Stream =" + defaultOutputStream);

		int numOfMappers = din.readInt();
		for (int i = 0; i < numOfMappers; i++) {
			String mapId = marshaller.readString(din);
			resultsTracker.put(mapId, 0);
		}
	}

	@Override
	public byte[] getBytes() throws IOException {
		byte[] marshalledBytes = null;
		MarshallingUtil marshaller = MarshallingUtil.getInstance();
		ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
		DataOutputStream dout = new DataOutputStream(new BufferedOutputStream(
				baOutputStream));

		initializeDatasetCollection();
		byte[] baseClassBytes = super.getBytes();
		dout.writeInt(baseClassBytes.length);
		dout.write(baseClassBytes);

		marshaller.writeString(dout, defaultOutputStream);

		dout.writeInt(mappers.size());
		Enumeration<String> mapIds = mappers.keys();
		while (mapIds.hasMoreElements()) {
			marshaller.writeString(dout, mapIds.nextElement());
		}

		dout.flush();
		marshalledBytes = baOutputStream.toByteArray();
		return marshalledBytes;

	}

	private void throwMapReduceException(String errorReport)
			throws MapReduceException {
		throw new MapReduceException(errorReport);
	}
}
